/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2013 All Rights Reserved.
 */
package com.pingan.mobilegw.biz.shared.router.executor;

import com.pingan.mobilegw.adapterservice.ResultEnum;
import com.pingan.mobilegw.adapterservice.model.MapOperation;
import com.pingan.mobilegw.core.model.GatewayRequest;
import com.pingan.mobilegw.core.model.GatewayResponse;
import com.pingan.mobilegw.core.model.enums.RequestSceneEnum;
import com.pingan.mobilegw.core.model.exception.MgwException;
import com.pingan.mobilegw.core.model.request.SingleRequestBody;
import com.pingan.mobilegw.core.model.response.SingleResponseBody;
import com.pingan.mobilegw.core.service.business.BusinessComponent;
import com.pingan.mobilegw.core.service.business.Processor;
import com.pingan.mobilegw.core.service.util.RequestUtil;
import com.pingan.mobilegw.core.service.util.ResponseUtil;
import com.pingan.mobilegw.mobileservice.invoke.MobileRpcHolder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 并行执行器
 * 
 * @author jiadao
 * @version $Id: ParallelExecutor.java, v 0.1 2013-5-14 下午3:14:41 jiadao Exp $
 */
public class ParallelBatchExecutor implements BatchExecutor {

    /** 默认logger */
    private static final Log logger = LogFactory.getLog(ParallelBatchExecutor.class);

    /** 线程池 */
    private Executor            multiThreadPoolExecutor;

    /** 服务容器组件 */
    private BusinessComponent   businessComponent;

    /** 安全管理器 */
    private SecurityManager     securityManager;

    public void execute(GatewayRequest gatewayRequest, GatewayResponse gatewayResponse) {

        // 将一个客户端请求分解为多个内部请求
        List<GatewayRequest> singleRequests = RequestUtil.convertBatchToSingle(gatewayRequest);

        BatchResponseRender responseRender = new BatchResponseRender(
            gatewayRequest.getHttpResponse(), singleRequests);

        // 将任务放入队列中
        CompletionService<SingleResponse> completionService = new ExecutorCompletionService<SingleResponse>(
            multiThreadPoolExecutor);
        for (GatewayRequest request : singleRequests) {
            completionService.submit(new ServiceRunnable(request));
        }

        for (int i = 0; i < singleRequests.size(); i++) {
            try {
                SingleResponse response = completionService.take().get();
                responseRender.flushResponseBody(response);
            } catch (InterruptedException e) {
                logger.error("线程池出错了", e);
            } catch (ExecutionException e) {
                logger.error("线程池出错了", e);
            }
        }

        responseRender.endResponseBody();
    }

    /**
     * 单任务服务线程
     * 
     * @author jiadao
     * @version $Id: MultiServiceRouterImpl.java, v 0.1 2013-4-23 下午4:15:21 jiadao Exp $
     */
    public class ServiceRunnable implements Callable<SingleResponse> {

        /** 网关请求 */
        private final GatewayRequest request;

        /**
         * 构造函数
         * 
         * @param request
         */
        private ServiceRunnable(GatewayRequest request) {
            this.request = request;
        }

        /**
         * 
         * @see java.util.concurrent.Callable#call()
         */
        @Override
        public SingleResponse call() throws Exception {
            GatewayResponse response = null;
            SingleRequestBody requestBody = (SingleRequestBody) request.getRequestBody();
            try {

                // 前置操作，各种检查。
               //TODO securityManager.check(request, RequestSceneEnum.SINGLE);

                Processor processor = businessComponent
                    .getProcessor(requestBody.getOperationType());

                response = processor.process(request);

            } catch (MgwException e) {
                logger.warn("无线网关路由错误" , e);
                response = ResponseUtil.buildErrorResponse(e);
            } catch (Exception e) {
                logger.error("服务路由未知异常", e);
                response = ResponseUtil.buildErrorResponse(ResultEnum.UnknowError);
            }

            ((SingleResponseBody) response.getResponseBody()).setId(requestBody.getId());

            SingleResponse singleResponse = new SingleResponse(response,
                MobileRpcHolder.getIncrementalChanges());

            logger.info("multi_resp1:" + singleResponse);

            return singleResponse;
        }
    }

    /**
     * 响应
     * 
     * @author jiadao
     * @version $Id: MultiServiceRouterImpl.java, v 0.1 2013-4-23 下午4:26:33 jiadao Exp $
     */
    public class SingleResponse {

        /** 网关响应 */
        private final GatewayResponse           response;

        /** 增量变化 */
        private final Map<String, MapOperation> incrementalChanges;

        /**
         * 构造函数
         * @param response
         */
        private SingleResponse(GatewayResponse response,
                               Map<String, MapOperation> incrementalChanges) {
            this.response = response;
            this.incrementalChanges = incrementalChanges;
        }

        /**
         * Getter method for property <tt>response</tt>.
         * 
         * @return property value of response
         */
        public GatewayResponse getResponse() {
            return response;
        }

        /**
         * Getter method for property <tt>incrementalChanges</tt>.
         * 
         * @return property value of incrementalChanges
         */
        public Map<String, MapOperation> getIncrementalChanges() {
            return incrementalChanges;
        }

    }

    /**
     * Setter method for property <tt>businessComponent</tt>.
     * 
     * @param businessComponent value to be assigned to property businessComponent
     */
    public void setBusinessComponent(BusinessComponent businessComponent) {
        this.businessComponent = businessComponent;
    }

    /**
     * Setter method for property <tt>multiThreadPoolExecutor</tt>.
     * 
     * @param multiThreadPoolExecutor value to be assigned to property multiThreadPoolExecutor
     */
    public void setMultiThreadPoolExecutor(Executor multiThreadPoolExecutor) {
        this.multiThreadPoolExecutor = multiThreadPoolExecutor;
    }

    /**
     * Setter method for property <tt>securityManager</tt>.
     * 
     * @param securityManager value to be assigned to property securityManager
     */
    public void setSecurityManager(SecurityManager securityManager) {
        this.securityManager = securityManager;
    }

}
